package net.wicp.tams.common.binlog.dump.handlerConsumer;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;

import com.lmax.disruptor.WorkHandler;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.apiext.LoggerUtil;
import net.wicp.tams.common.binlog.dump.MainDump;
import net.wicp.tams.common.binlog.dump.bean.Dump;
import net.wicp.tams.common.binlog.dump.bean.DumpEvent;
import net.wicp.tams.common.binlog.dump.listener.IBusiSender;
import net.wicp.tams.common.constant.JvmStatus;

@Slf4j
public class SendHander implements WorkHandler<DumpEvent> {
	private final Map<String, IBusiSender<DumpEvent>> sendmap = new HashMap<String, IBusiSender<DumpEvent>>();
	private final Map<String, Dump> dumpmap = new HashMap<String, Dump>();
	private final String formatestr = "%s.%s";

	public SendHander(Dump[] dumps) {
		for (Dump dump : dumps) {
			sendmap.put(String.format(formatestr, dump.getDb(), dump.getTb()), dump.getBusiSender());
			dumpmap.put(String.format(formatestr, dump.getDb(), dump.getTb()), dump);
		}
	}

	@Override
	public void onEvent(DumpEvent event) throws Exception {
		Thread.currentThread().setName("SendHanderThread");
		String key = String.format(formatestr, event.getDump().getDb(), event.getDump().getTb());
		IBusiSender<DumpEvent> iBusiSend = this.sendmap.get(key);
		if (iBusiSend == null || event == null || CollectionUtils.isEmpty(event.getDatas())) {
			this.dumpmap.get(key).getMetric().counter_send_num.inc(event.getDatas().size());
			this.dumpmap.get(key).getMetric().counter_send_event.inc();
			this.dumpmap.get(key).setLastId(event.getEndId());
			isOver();
			return;
		}
		try {
			iBusiSend.doSend(event);
			this.dumpmap.get(key).getMetric().counter_send_num.inc(event.getDatas().size());
			this.dumpmap.get(key).getMetric().counter_send_event.inc();
			this.dumpmap.get(key).setLastId(event.getEndId());
		} catch (Throwable e) {
			log.error("业务处理失败", e);// TODO
			LoggerUtil.exit(JvmStatus.s15);
		} finally {
			// log.info("sendTime:{},num:{}", (System.currentTimeMillis() -
			// curTime),event.getEsDataBuilder().getDatasList().size());
			isOver();
		}
	}

	private void isOver() {
		boolean isover = true;
		for (Publisher publisher : MainDump.publishers) {
			if (!publisher.isIsover()) {
				isover = false;
				break;
			}
		}
		if (isover) {
			log.info("dump sucess!");
			LoggerUtil.exit(JvmStatus.s15);
		}
	}
}
